# Copyright (c) Streamlit Inc. (2018-2022) Snowflake Inc. (2022-2025)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import re

import pytest
from playwright.sync_api import Locator, Page, expect

from e2e_playwright.conftest import (
    ImageCompareFunction,
    wait_until,
)
from e2e_playwright.shared.app_utils import (
    check_top_level_class,
    click_button,
    click_checkbox,
    goto_app,
    select_radio_option,
)

VIDEO_ELEMENTS_COUNT = 12


def _select_video_to_show(app: Page, label: str) -> Locator:
    select_radio_option(app, re.compile(f"^{label}$"))
    video_element = app.get_by_test_id("stVideo").first
    # Prevent flakiness: we move the mouse before scrolling to prevent the cursor
    # hovering over a video element and, thereby, changing how the video interface is
    # rendered (e.g. without the controls in the bottom which are hidden)
    app.mouse.move(0, 0)
    video_element.scroll_into_view_if_needed()
    expect(video_element).to_be_visible()
    return video_element


def _wait_until_video_has_data(app: Page, video_element: Locator):
    # To prevent flakiness, we wait for the video to load and start playing
    # The readyState is defined in https://developer.mozilla.org/en-US/docs/Web/API/HTMLMediaElement/readyState
    # 3 means there is some data to play now and few frames for the future. On webkit
    # this seems to be flaky, so we check also the duration of the video.
    wait_until(
        app,
        lambda: video_element.evaluate("el => el.readyState >= 3 || el.duration > 0")
        is True,
        timeout=15000,
    )
    # Wait another 2 seconds to prevent some flakiness
    app.wait_for_timeout(2000)


@pytest.mark.skip_browser("webkit")
def test_video_width_configurations(app: Page, assert_snapshot: ImageCompareFunction):
    """Test that `st.video` width configurations are applied correctly."""
    video_element = _select_video_to_show(app, "webm video with pixel width")
    _wait_until_video_has_data(app, video_element)
    assert_snapshot(video_element, name="st_video-width_400px", image_threshold=0.1)

    video_element = _select_video_to_show(app, "webm video with stretch width")
    _wait_until_video_has_data(app, video_element)
    assert_snapshot(video_element, name="st_video-width_stretch", image_threshold=0.1)


# Chromium miss codecs required to play that mp3 videos
# https://www.howtogeek.com/202825/what%E2%80%99s-the-difference-between-chromium-and-chrome/
@pytest.mark.skip_browser("chromium")
def test_video_rendering(app: Page, assert_snapshot: ImageCompareFunction):
    """Test that `st.video` renders correctly via screenshots matching."""

    video_element = _select_video_to_show(app, "mp4 video")
    _wait_until_video_has_data(app, video_element)
    assert_snapshot(
        video_element,
        name="video_element_first",
        image_threshold=0.1,
    )

    video_element = _select_video_to_show(app, "mp4 video with subtitles")
    _wait_until_video_has_data(app, video_element)
    assert_snapshot(
        video_element,
        name="video_element_with_subtitles",
        image_threshold=0.1,
    )


@pytest.mark.skip_browser("webkit")
def test_video_rendering_webm(app: Page, assert_snapshot: ImageCompareFunction):
    """Test that `st.video` renders correctly webm video via screenshots matching."""

    video_element = _select_video_to_show(app, "webm video with subtitles")
    _wait_until_video_has_data(app, video_element)

    assert_snapshot(
        video_element,
        name="video_element_webm_with_subtitles",
        image_threshold=0.1,
    )


def test_displays_a_video_player(app: Page):
    video_element = _select_video_to_show(app, "mp4 video")
    # src here is a generated by streamlit url since we pass a file content
    expect(video_element).to_have_attribute("src", re.compile(r".*media.*.mp4"))


@pytest.mark.flaky(reruns=3)
@pytest.mark.parametrize(
    "video_option_label",
    [
        pytest.param(
            "webm video with end time", marks=pytest.mark.skip_browser("webkit")
        ),
        pytest.param(
            "mp4 video with end time", marks=pytest.mark.skip_browser("chromium")
        ),
    ],
)
def test_video_end_time(app: Page, video_option_label: str):
    """Test that `st.video` with end_time works correctly."""

    video_element = _select_video_to_show(app, video_option_label)
    _wait_until_video_has_data(app, video_element)
    # Give the video a little more time to load
    # And reduce potential flakiness:
    app.wait_for_timeout(2000)
    video_element.evaluate("el => el.play()")

    # Wait for the video to actually start playing
    wait_until(
        app,
        lambda: video_element.evaluate("el => !el.paused") is True,
        timeout=5000,
    )

    # Wait until video reaches end_time and pauses
    wait_until(
        app,
        lambda: video_element.evaluate("el => el.paused") is True,
        timeout=10000,
    )

    # Verify the video stopped at the expected end_time (33 seconds)
    wait_until(app, lambda: int(video_element.evaluate("el => el.currentTime")) == 33)


@pytest.mark.parametrize(
    "video_option_label",
    [
        pytest.param(
            "webm video with end time and loop",
            marks=pytest.mark.skip_browser("webkit"),
        ),
        pytest.param(
            "mp4 video with end time and loop",
            marks=pytest.mark.skip_browser("chromium"),
        ),
    ],
)
def test_video_end_time_loop(app: Page, video_option_label: str):
    """Test that `st.video` with end_time and loop works correctly."""
    video_element = _select_video_to_show(app, video_option_label)
    _wait_until_video_has_data(app, video_element)

    video_element.evaluate("el => el.play()")
    # According to the element definition looks like this:
    # start_time=35, end_time=39, loop=True
    # We wait for 6 seconds, which mean the current time should be approximately 37:
    # 4 seconds until end_time and 2 seconds starting from start time
    app.wait_for_timeout(6000)
    expect(video_element).to_have_js_property("paused", False)
    wait_until(app, lambda: 36 < video_element.evaluate("el => el.currentTime") < 38)


def test_video_autoplay(app: Page):
    """Test that `st.video` autoplay property works correctly."""
    video_element = _select_video_to_show(app, "webm video with autoplay")
    expect(video_element).to_have_js_property("paused", True)
    expect(video_element).to_have_js_property("autoplay", False)

    click_checkbox(app, "Autoplay")

    _wait_until_video_has_data(app, video_element)
    expect(video_element).to_have_js_property("autoplay", True)
    expect(video_element).to_have_js_property("paused", False)


@pytest.mark.skip_browser("webkit")  # Flakiness with this in CI
def test_video_muted_autoplay(app: Page):
    """Test that `st.video` muted and autoplay properties work correctly."""
    video_element = _select_video_to_show(app, "webm video muted")
    _wait_until_video_has_data(app, video_element)

    expect(video_element).to_have_js_property("muted", True)
    expect(video_element).to_have_js_property("autoplay", True)
    expect(video_element).to_have_js_property("paused", False)


def test_video_remount_no_autoplay(app: Page):
    """Test that `st.video` remounts correctly without autoplay."""
    video_element = _select_video_to_show(app, "webm video with autoplay")
    _wait_until_video_has_data(app, video_element)

    expect(video_element).to_have_js_property("paused", True)
    expect(video_element).to_have_js_property("autoplay", False)

    click_checkbox(app, "Autoplay")

    expect(video_element).to_have_js_property("autoplay", True)
    expect(video_element).to_have_js_property("paused", False)

    click_checkbox(app, "Autoplay")
    click_button(app, "Create some elements to unmount component")

    expect(video_element).to_have_js_property("autoplay", False)
    expect(video_element).to_have_js_property("paused", True)


def test_check_top_level_class(app: Page):
    """Check that the top level class is correctly set."""
    _select_video_to_show(app, "webm video with autoplay")
    check_top_level_class(app, "stVideo")


def test_video_source_error(app: Page, app_port: int):
    """Test `st.video` source error."""
    # Ensure video source request return a 404 status
    app.route(
        f"http://localhost:{app_port}/media/**",
        lambda route: route.fulfill(
            status=404, headers={"Content-Type": "text/plain"}, body="Not Found"
        ),
    )

    # Capture console messages
    messages = []
    app.on("console", lambda msg: messages.append(msg.text))

    # Navigate to the app
    goto_app(app, f"http://localhost:{app_port}")
    _select_video_to_show(app, "mp4 video")

    # Wait until the expected error is logged, indicating CLIENT_ERROR was sent
    wait_until(
        app,
        lambda: any(
            "Client Error: Video source error" in message for message in messages
        ),
    )
